package org.egomsl.mw;

import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Writable;
import org.egomsl.mw.record.Record;
import org.egomsl.mw.record.RecordImpl;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.nio.ByteBuffer;

public class HadoopRecord implements Writable {
    private Record record;

    public HadoopRecord() {
        record = new RecordImpl();
    }

    public HadoopRecord(Record record) {
        this.record = record;
    }

    public Record getRecord() {
        return record;
    }

    public BytesWritable getKey() {
        BytesWritable key = new BytesWritable();
        if(record != null) {
            key.set(record.getKey().getKeyBytes(),
                    record.getKey().getKeyOffset(),
                    record.getKey().getKeyLen());
        }
        return key;
    }

    @Override
    public void write(DataOutput out) throws IOException {
        int size = record.getDataLen();
        byte[] dataBytes = new byte[size];
        ByteBuffer dataBuffer = ByteBuffer.wrap(dataBytes);
        record.writeData(dataBuffer);
        out.write(dataBytes);
    }

    @Override
    public void readFields(DataInput in) throws IOException {
        int size = in.readInt();
        ByteBuffer buffer = ByteBuffer.allocate(size);
        buffer.putInt(size);
        in.readFully(buffer.array(), buffer.position(), buffer.limit());
        buffer.flip();
    }
}
